/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2024-2024. All rights reserved.
 * Description: typical send/recv
 */

#include "send_recv_executor.h"
#include "device_capacity.h"
#include "dispatcher_pub.h"
#include "adapter_rts.h"
#include "typical_mr_manager.h"
#include "externalinput_pub.h"

namespace hccl {
constexpr u32 HCCL_POLL_CQ_DEPTH = 32;
constexpr u32 HCCL_POLL_CQ_ONETIME = 1;
constexpr u32 HCCL_POLL_CQ_INTERVAL = 100;
SendRecvExecutor::SendRecvExecutor(HcclRtStream stream, QpHandle qpHandle,
    const struct mr_info& localWindowMem, const struct mr_info& remoteWindowMem,
    const struct mr_info& localSyncMemPrepare, const struct mr_info& localSyncMemDone, const struct mr_info& localSyncMemAck,
    const struct mr_info& remoteSyncMemPrepare, const struct mr_info& remoteSyncMemDone, const struct mr_info& remoteSyncMemAck,
    u32 immData, const u64 chunkNum)
    : stream_(stream), qpHandle_(qpHandle), localWindowMem_(localWindowMem), remoteWindowMem_(remoteWindowMem),
    localSyncMemPrepare_(localSyncMemPrepare), localSyncMemDone_(localSyncMemDone), localSyncMemAck_(localSyncMemAck),
    remoteSyncMemPrepare_(remoteSyncMemPrepare), remoteSyncMemDone_(remoteSyncMemDone),
    remoteSyncMemAck_(remoteSyncMemAck), immData_(immData), chunkSize_(chunkNum), 
    notifyWaitMode_(SyncMode::DEFAULT_TIMEWAITSYNCMODE)
{
}

SendRecvExecutor::~SendRecvExecutor()
{}

HcclResult SendRecvExecutor::Init()
{
    CHK_RET(hrtGetNotifySize(notifySize_));
    if (GetExternalInputHcclExecTimeoutSet() != HcclExecTimeoutSet::HCCL_EXEC_TIMEOUT_NOT_SET) {
        notifyWaitMode_ = SyncMode::CONFIGURABLE_TIMEWAITSYNCMODE;
    }
    CHK_RET(TypicalSyncMem::GetInstance().GetNotifyHandle(reinterpret_cast<u64>(localSyncMemPrepare_.addr),
        prepareNotify_));
    CHK_PTR_NULL(prepareNotify_);

    CHK_RET(TypicalSyncMem::GetInstance().GetNotifyHandle(reinterpret_cast<u64>(localSyncMemAck_.addr),
        ackNotify_));
    CHK_PTR_NULL(ackNotify_);

    CHK_RET(TypicalSyncMem::GetInstance().GetNotifyHandle(reinterpret_cast<u64>(localSyncMemDone_.addr),
        doneNotify_));
    CHK_PTR_NULL(doneNotify_);

    CHK_RET(TypicalSyncMem::GetInstance().GetNotifySrcMem(notifySrcMem_));
    CHK_PTR_NULL(notifySrcMem_.addr);

    HCCL_INFO("[SendRecvExecutor][Init] SendRecvExecutor init success! notifySize[%u], notifyWaitMode[%d], "\
        "prepareNotify[%p], ackNotify[%p], doneNotify[%p], notifySrcMem addr[%p], localWindowMem addr[%p], "\
        "remoteWindowMem addr[%p], remoteSyncMemPrepare addr[%p], remoteSyncMemDone addr[%p], "\
        "remoteSyncMemAck addr[%p], immData[%u]",
        notifySize_, notifyWaitMode_, prepareNotify_, ackNotify_, doneNotify_,
        notifySrcMem_.addr, localWindowMem_.addr, remoteWindowMem_.addr,
        remoteSyncMemPrepare_.addr, remoteSyncMemDone_.addr, remoteSyncMemAck_.addr, immData_);
    return HCCL_SUCCESS;
}

HcclResult SendRecvExecutor::IsOverlappedWithWinMem(void* userPtr, u64 userMemSize, bool& isOverlapped)
{
    if (userPtr >= localWindowMem_.addr &&
        static_cast<u8*>(userPtr) + userMemSize <= static_cast<u8*>(localWindowMem_.addr) + localWindowMem_.size) {
        isOverlapped = true;
    } else if (static_cast<u8*>(userPtr) + userMemSize <= static_cast<u8*>(localWindowMem_.addr) ||
        static_cast<u8*>(userPtr) >= static_cast<u8*>(localWindowMem_.addr) + localWindowMem_.size){
        isOverlapped = false;
    } else {
        HCCL_ERROR("[SendRecvExecutor][IsOverlappedWithWinMem] The user mem addr or size is illegal. "\
        "The addr of user mem is %p, user mem size is %llu. The addr of window mem is %p, window mem size is %llu.",
        userPtr, userMemSize, localWindowMem_.addr, localWindowMem_.size);
        return HCCL_E_PARA;
    }
    return HCCL_SUCCESS;
}

HcclResult SendRecvExecutor::Send(void* inputPtr, u64 count, HcclDataType dataType)
{
    HcclResult ret = HCCL_SUCCESS;
    void *windowsMemPtr = reinterpret_cast<void*>(localWindowMem_.addr);
    uint32_t unitSize = SIZE_TABLE[dataType];
    uint8_t *curInputPtr = static_cast<uint8_t *>(inputPtr);
    CHK_PTR_NULL(curInputPtr);
    uint64_t inputOffset = 0;
    uint64_t countLeft = count;
    Stream streamObj(stream_);
    // 判断userMem是否是windowMem的一部分
    bool isOverlapped = false;
    CHK_RET(IsOverlappedWithWinMem(inputPtr, count * unitSize, isOverlapped));

    u64 maxCountPerLoop = localWindowMem_.size / unitSize;

    while (countLeft > 0) {
        // 防止数据回绕
        CHK_PRT_RET(countLeft > count, HCCL_ERROR("[SendRecvExecutor][Send] countLeft is underflow."),
            HCCL_E_PARA);
        curInputPtr += inputOffset;
        HCCL_DEBUG("[SendRecvExecutor]][Send] InputOffset[%llu]", inputOffset);
        u64 curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
        u64 curSize = curCount * unitSize;
        HCCL_DEBUG("[SendRecvExecutor][Send] curInputPtr[%p], curCount[%llu], curSize[%llu]", curInputPtr,
            curCount, curSize);
        DeviceMem inMem(curInputPtr, curSize);
        // 如果userMem是否是windowMem的一部分，跳过D2D拷
        if (isOverlapped) {
            ret = SendRun(inMem);
        } else {
            DeviceMem inWindowMem(windowsMemPtr, curSize);
            CHK_RET(MemcpyAsyncD2D(inWindowMem, inMem, streamObj));
            ret = SendRun(inWindowMem);
        }
        CHK_PRT_RET(ret != HCCL_SUCCESS,
            HCCL_ERROR("[SendRecvExecutor][Send] errNo[0x%016llx] send error, ptr[%p], count[%llu], dataType[%d]",
            HCCL_ERROR_CODE(ret), windowsMemPtr, curCount, dataType), ret);
        CHK_PRT_RET((curCount == 0), HCCL_ERROR("[SendRecvExecutor]][Send]In OP_BASE curCount is zero"), HCCL_E_PARA);
        countLeft -= curCount;
        inputOffset = curSize;
    }
    return HCCL_SUCCESS;
}

HcclResult SendRecvExecutor::Receive(void* outputPtr, u64 count, HcclDataType dataType)
{
    HcclResult ret = HCCL_SUCCESS;
    void *windowsMemPtr = reinterpret_cast<void*>(localWindowMem_.addr);
    uint32_t unitSize = SIZE_TABLE[dataType];
    uint8_t *curOutPutPtr = static_cast<uint8_t *>(outputPtr);
    CHK_PTR_NULL(curOutPutPtr);
    uint64_t outputOffset = 0;
    uint64_t countLeft = count;
    Stream streamObj(stream_);
    // 判断userMem是否是windowMem的一部分
    bool isOverlapped = false;
    CHK_RET(IsOverlappedWithWinMem(outputPtr, count * unitSize, isOverlapped));

    u64 maxCountPerLoop = localWindowMem_.size / unitSize;

    while (countLeft > 0) {
        // 防止数据回绕
        CHK_PRT_RET(countLeft > count, HCCL_ERROR("[SendRecvExecutor][Receive] countLeft is underflow."),
            HCCL_E_PARA);
        curOutPutPtr += outputOffset;
        HCCL_INFO("[SendRecvExecutor][Receive] inputOffset[%llu]", outputOffset);
        u64 curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
        u64 curSize = curCount * unitSize; // 单位 byte
        HCCL_INFO("[SendRecvExecutor][Receive] curOutPutPtr[%p], curCount[%llu], curSize[%llu]", curOutPutPtr,
            curCount, curSize);
        DeviceMem outMem(curOutPutPtr, curSize);
        // 如果userMem是否是windowMem的一部分，跳过D2D拷贝
        if (isOverlapped) {
            if (immData_ != 0) {
                ret = ReceiveRunByPollCq(outMem);
            } else {
                ret = ReceiveRun(outMem);
            }
            CHK_PRT_RET(ret != HCCL_SUCCESS,
                HCCL_ERROR("[SendRecvExecutor][Receive] errNo[0x%016llx] Receive error, isOverlapped[%d], ptr[%p], count[%llu], dataType[%d]",
                HCCL_ERROR_CODE(ret), isOverlapped, windowsMemPtr, curCount, dataType), ret);
        } else {
            DeviceMem outWindowMem(windowsMemPtr, curSize);
            if (immData_ != 0) {
                ret = ReceiveRunByPollCq(outWindowMem);
            } else {
                ret = ReceiveRun(outWindowMem);
            }
            CHK_PRT_RET(ret != HCCL_SUCCESS,
                HCCL_ERROR("[SendRecvExecutor][Receive] errNo[0x%016llx] Receive error, isOverlapped[%d], ptr[%p], count[%llu], dataType[%d]",
                HCCL_ERROR_CODE(ret), isOverlapped, windowsMemPtr, curCount, dataType), ret);
            CHK_RET(MemcpyAsyncD2D(outMem, outWindowMem, streamObj));
        }
        CHK_PRT_RET((curCount == 0), HCCL_ERROR("[SendRecvExecutor][Receive]In OP_BASE curCount is zero"), HCCL_E_PARA);
        countLeft -= curCount;
        outputOffset = curSize;
    }
    return HCCL_SUCCESS;
}

HcclResult SendRecvExecutor::SendRun(DeviceMem& sendBuffer)
{
    HcclResult ret = HCCL_SUCCESS;
    if (!sendBuffer) {
        HCCL_ERROR("[SendRecvExecutor][SendRun] Send buffer ptr is null.");
        return HCCL_E_PTR;
    }
    u64 sizePerRound = 0;
    u64 sizePerSlice = chunkSize_;
    u64 length = sendBuffer.size();
    u64 offset = 0;

    for (u64 sizeResidue = length; sizeResidue > 0; sizeResidue -= sizePerRound) {
        // 防止数据回绕
        CHK_PRT_RET(sizeResidue > length, HCCL_ERROR("[SendRecvExecutor][SendRun] countLeft is underflow."),
            HCCL_E_PARA);
        ret = WaitSignal(prepareNotify_);
        CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][SendRun] Wait prepare failed"), ret);
        offset += sizePerRound;
        sizePerRound = (sizeResidue > sizePerSlice) ? sizePerSlice : sizeResidue;
        void* localAddr = static_cast<u8 *>(localWindowMem_.addr) + offset;
        HCCL_INFO("rx async inputmem's offset[%llu] size[%llu]", offset, sizePerRound);

        ret = PayLoad(localAddr, offset, sizePerRound);
        CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][SendRun] Send data fail with offset[%llu] "\
            "size[%llu] failed", offset, sizePerRound), ret);

        if(immData_ == 0) {
            ret = RecordNotify(remoteSyncMemDone_.addr, remoteSyncMemDone_.lkey,
                notifySrcMem_.addr, notifySrcMem_.lkey, notifySize_);
            CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][SendRun] Record done failed"), ret);
        }

        ret = WaitSignal(ackNotify_);
        CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][SendRun] Wait ack failed"), ret);
    }
    return HCCL_SUCCESS;
}

HcclResult SendRecvExecutor::PollCq()
{
    auto startTime = std::chrono::steady_clock::now();
    auto timeout = std::chrono::seconds(GetExternalInputHcclExecTimeOut());

    struct ibv_wc wc[HCCL_POLL_CQ_DEPTH];
    while ((std::chrono::steady_clock::now() - startTime) < timeout) {
        s32 num = hrtRaPollCq(qpHandle_, false, HCCL_POLL_CQ_ONETIME, wc);
        if (static_cast<u32>(num) < HCCL_POLL_CQ_ONETIME) {
            SaluSleep(HCCL_POLL_CQ_INTERVAL);
            continue;
        }
        for (int i = 0; i < num; i++) {
            if (wc[i].status != 0) {
                HCCL_ERROR("rdma poll tag sq failed, cqe status[%u]", wc[i].status);
                return HCCL_E_INTERNAL;
            }
        }
        return HCCL_SUCCESS;
    }
    HCCL_ERROR("[SendRecvExecutor][PollCq] Wait Cqe timeOut[%d] s", GetExternalInputHcclLinkTimeOut());
    return HCCL_E_TIMEOUT;
}

HcclResult SendRecvExecutor::ReceiveRunByPollCq(DeviceMem& receiveBuffer)
{
    HcclResult ret = HCCL_SUCCESS;
    if (!receiveBuffer) {
        HCCL_ERROR("[SendRecvExecutor][ReceiveRunByPollCq] Receive buffer ptr is null.");
        return HCCL_E_PTR;
    }
    u64 sizePerRound = 0;
    u64 sizePerSlice = chunkSize_;
    u64 length = receiveBuffer.size();
 
    u64 offset = 0;
 
    for (u64 sizeResidue = length; sizeResidue > 0; sizeResidue -= sizePerRound) {
        // 防止数据回绕
        CHK_PRT_RET(sizeResidue > length, HCCL_ERROR("[SendRecvExecutor][ReceiveRunByPollCq] countLeft is underflow."),
            HCCL_E_PARA);
        offset += sizePerRound;
        sizePerRound = (sizeResidue > sizePerSlice) ? sizePerSlice : sizeResidue;
        HCCL_INFO("rx async inputmem's offset[%llu] size[%llu]", offset, sizePerRound);
 
        void* localAddr = static_cast<u8 *>(localWindowMem_.addr) + offset;
        
        std::vector<struct recv_wrlist_data> recvWrVec(1);
        recvWrVec[0].wr_id = reinterpret_cast<u64>(localWindowMem_.addr);
        recvWrVec[0].mem_list.addr = reinterpret_cast<u64>(localAddr);
        recvWrVec[0].mem_list.len = sizePerRound;
        recvWrVec[0].mem_list.lkey = localWindowMem_.lkey;
 
        struct recv_wrlist_data *recvWr = recvWrVec.data();
        u32 completeNum = 0;
        ret = hrtRaRecvWrlist(qpHandle_, recvWr, 1, &completeNum);
        if (ret == HCCL_SUCCESS && completeNum == 1) {
            HCCL_INFO("[SendRecvExecutor][ReceiveRunByPollCq] Exec hrtRaRecvWrlist success.");
        } else {
            HCCL_ERROR("[SendRecvExecutor][ReceiveRunByPollCq] In RdmaDataTransport, hrtRaRecvWrlist failed. ret[%d], completeNum[%d].",
                ret, completeNum);
            return HCCL_E_NETWORK;
        }
 
        ret = RecordNotify(remoteSyncMemPrepare_.addr, remoteSyncMemPrepare_.lkey,
            notifySrcMem_.addr, notifySrcMem_.lkey, notifySize_);
        CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("SendRecvExecutor][ReceiveRunByPollCq] Record prepare failed"), ret);

        CHK_RET(PollCq());

        ret = RecordNotify(remoteSyncMemAck_.addr, remoteSyncMemAck_.lkey,
            notifySrcMem_.addr, notifySrcMem_.lkey, notifySize_);
        CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][ReceiveRunByPollCq] Record ack failed"), ret);
    }
    return HCCL_SUCCESS;
}

HcclResult SendRecvExecutor::ReceiveRun(DeviceMem& receiveBuffer)
{
    HcclResult ret = HCCL_SUCCESS;
    if (!receiveBuffer) {
        HCCL_ERROR("[SendRecvExecutor][ReceiveRun] Receive buffer ptr is null.");
        return HCCL_E_PTR;
    }
    u64 sizePerRound = 0;
    u64 sizePerSlice = chunkSize_;
    u64 length = receiveBuffer.size();
    u64 offset = 0;

    for (u64 sizeResidue = length; sizeResidue > 0; sizeResidue -= sizePerRound) {
        // 防止数据回绕
        CHK_PRT_RET(sizeResidue > length, HCCL_ERROR("[SendRecvExecutor][ReceiveRun] countLeft is underflow."),
            HCCL_E_PARA);
        ret = RecordNotify(remoteSyncMemPrepare_.addr, remoteSyncMemPrepare_.lkey,
            notifySrcMem_.addr, notifySrcMem_.lkey, notifySize_);
        CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("SendRecvExecutor][ReceiveRun] Record prepare failed"), ret);
        offset += sizePerRound;
        sizePerRound = (sizeResidue > sizePerSlice) ? sizePerSlice : sizeResidue;

        HCCL_INFO("[SendRecvExecutor][ReceiveRun]rx async inputmem's offset[%llu] size[%llu]", offset, sizePerRound);

        ret = WaitSignal(doneNotify_);
        CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][ReceiveRun] Wait done failed"), ret);

        ret = RecordNotify(remoteSyncMemAck_.addr, remoteSyncMemAck_.lkey,
            notifySrcMem_.addr, notifySrcMem_.lkey, notifySize_);
        CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][ReceiveRun] Wait ack failed"), ret);
    }
    return HCCL_SUCCESS;
}


HcclResult SendRecvExecutor::RecordNotify(void *dstMemPtr, u32 rkey, const void *srcMemPtr, u32 lkey, u64 srcMemSize)
{
    struct sg_list list = {0};
    struct send_wr_v2 wr = {0};
    // 构造wr信息
    list.addr = static_cast<u64>(reinterpret_cast<uintptr_t>(srcMemPtr));
    list.len = srcMemSize;
    list.lkey = lkey;

    wr.buf_list = &list;
    wr.buf_num = 1; /* 此处list只有一个，设置为1 */
    wr.dst_addr = static_cast<u64>(reinterpret_cast<uintptr_t>(dstMemPtr));
    wr.rkey = rkey;
    wr.op = RA_WR_RDMA_WRITE;
    wr.send_flag = RA_SEND_SIGNALED;

    HCCL_INFO("[SendRecvExecutor][RecordNotify] " \
        "Notify's dst addr[%p], local addr[%p], data's len[%u], remote mr key[%u], local mr key[%u]",
        wr.dst_addr, wr.buf_list->addr, wr.buf_list->len, wr.rkey, wr.buf_list->lkey);

    // RDMA异步发送
    CHK_RET(RdmaSendAsync(wr));
    return HCCL_SUCCESS;
}

HcclResult SendRecvExecutor::WaitSignal(HcclRtSignal signal)
{
    if (notifyWaitMode_ == SyncMode::CONFIGURABLE_TIMEWAITSYNCMODE) {
        CHK_RET(hrtNotifyWaitWithTimeOut(static_cast<HcclRtNotify>(signal), stream_,
            GetExternalInputHcclExecTimeOut()));
    } else {
        CHK_RET(hrtNotifyWaitWithTimeOut(static_cast<HcclRtNotify>(signal), stream_, NOTIFY_DEFAULT_WAIT_TIME));
    }

    u32 taskID = 0;
    u32 streamID = 0;
    hrtGetTaskIdAndStreamID(taskID, streamID);
    HCCL_INFO("[SendRecvExecutor][SignalWait] notifyId taskId[%u], streamID[%u]", taskID, streamID);
    return HCCL_SUCCESS;
}

HcclResult SendRecvExecutor::PayLoad(const void *src, u64 dstOffset, u64 len)
{
    HcclResult ret;
    HCCL_DEBUG("[SendRecvExecutor][PayLoad] Local window memory srcPtr[%p] len[%llu] dstOffset[%llu]",
        src, len, dstOffset);

    u32 txSendDataTimes = (len == 0) ? 1 : (len + RDMA_SEND_MAX_SIZE - 1) / RDMA_SEND_MAX_SIZE;

    for (u32 txSendDataIdx = 0; txSendDataIdx < txSendDataTimes; txSendDataIdx++) {
        u64 txSendDataOffset = txSendDataIdx * RDMA_SEND_MAX_SIZE;
        u64 txSendDataSize = (txSendDataIdx == (txSendDataTimes - 1)) ? len - txSendDataOffset : RDMA_SEND_MAX_SIZE;

        void* txdstMemPtr = reinterpret_cast<void *>(reinterpret_cast<u8*>(remoteWindowMem_.addr) + dstOffset +
            txSendDataOffset);

        const void* txsrcMemPtr = reinterpret_cast<const void *>(reinterpret_cast<const char *>(src) +
            txSendDataOffset);
        struct send_wr_v2 wr{};
        // 构造wr信息
        wr.buf_num = 1; /* 此处list只有一个，设置为1 */
        wr.dst_addr = static_cast<u64>(reinterpret_cast<uintptr_t>(txdstMemPtr));
        wr.rkey = remoteWindowMem_.lkey;
        wr.send_flag = RA_SEND_SIGNALED;
        if(immData_ != 0) {
            wr.op = RA_WR_RDMA_WRITE_WITH_IMM;
            wr.ext.imm_data = immData_;
        } else {
            wr.op = RA_WR_RDMA_WRITE;
        }
        struct sg_list list = {0};
        list.addr = static_cast<u64>(reinterpret_cast<uintptr_t>(txsrcMemPtr));
        list.len = txSendDataSize;
        list.lkey = localWindowMem_.lkey;
        wr.buf_list = &list;
        ret = RdmaSendAsync(wr);
        CHK_PRT_RET(ret != HCCL_SUCCESS,
            HCCL_ERROR("[SendRecvExecutor][PayLoad]errNo[0x%016llx] In lbv exp, add wqe list failed."\
                "srcMemSize[%llu]", HCCL_ERROR_CODE(ret), txSendDataSize), ret);
    }

    return HCCL_SUCCESS;
}

HcclResult SendRecvExecutor::RdmaSendAsync(struct send_wr_v2 &wr)
{
    HcclResult ret = HCCL_SUCCESS;
    struct send_wr_rsp opRsp = {0};
    HCCL_DEBUG("[SendRecvExecutor][RdmaSendAsync] dst_addr[%p], src_addr[%p], len[%u]",
        wr.dst_addr, wr.buf_list->addr, wr.buf_list->len);

    CHK_RET(HrtRaSendWrV2(qpHandle_, &wr, &opRsp, GetWorkflowMode()));

    u32 dbIndex = static_cast<u32>(opRsp.db.db_index);
    u64 dbInfo = static_cast<u64>(opRsp.db.db_info);

    if ((dbIndex == INVALID_UINT) && (dbInfo == INVALID_U64)) {
        // zero byte message 不需要下发rdma send task
        HCCL_DEBUG("[SendRecvExecutor][RdmaSendAsync] dbIndex and dbInfo is invalid.");
        return HCCL_SUCCESS;
    }

    ret = hrtRDMADBSend(dbIndex, dbInfo, stream_);
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[SendRecvExecutor][RdmaSendAsync]errNo[0x%016llx] In lbv exp op base mode, "\
        "rdma send failed. dbIndex[%u] dbInfo[%llu]", HCCL_ERROR_CODE(ret), dbIndex, dbInfo), ret);

    u32 taskID = 0;
    u32 streamID = 0;
    hrtGetTaskIdAndStreamID(taskID, streamID);
    HCCL_INFO("[SendRecvExecutor][RdmaSendAsync] dbIndex[%u], dbInfo[%llu], taskID[%u], streamID[%u]",
        dbIndex, dbInfo, taskID, streamID);
    return HCCL_SUCCESS;
}

HcclResult SendRecvExecutor::MemcpyAsyncD2D(hccl::DeviceMem &dst, const hccl::DeviceMem &src, hccl::Stream &stream)
{
    CHK_PTR_NULL(dst.ptr());
    CHK_PTR_NULL(src.ptr());

    if (stream.ptr() == nullptr) {
        CHK_SAFETY_FUNC_RET(memcpy_s(dst.ptr(), dst.size(), src.ptr(), src.size()));
        return HCCL_E_PARA;
    }

    if (src.size() == 0) {
        HCCL_DEBUG("[SendRecvExecutor][MemcpyAsyncD2D] count is 0, return success.");
        return HCCL_SUCCESS;
    }

    uint64_t spiltLoop = 0;
    uint64_t addrOffset = 0;
    uint64_t contSplit = 0;
    if (src.size() > HCCL_SDMA_MAX_COUNT_4GB) {
        spiltLoop = (src.size() % HCCL_SDMA_MAX_COUNT_4GB) ?
            (src.size() / HCCL_SDMA_MAX_COUNT_4GB) : ((src.size() / HCCL_SDMA_MAX_COUNT_4GB) - 1);
        HCCL_INFO("[SendRecvExecutor][MemcpyAsyncD2D] MemcpyAsync SDMA task countSize is bigger than 4GB "\
            "and do segmentation splitloop[%llu]", spiltLoop);
    }
    /* SDMA任务拆分 */
    for (uint64_t index = 0 ; index <= spiltLoop; index++) {
        addrOffset = index * HCCL_SDMA_MAX_COUNT_4GB;
        contSplit = (index == spiltLoop) ? (src.size() - index * HCCL_SDMA_MAX_COUNT_4GB) : (HCCL_SDMA_MAX_COUNT_4GB);
        void *srcSplit = static_cast<void *>(static_cast<u8*>(const_cast<void*>(src.ptr())) + addrOffset);
        void *dstSplit = static_cast<void *>(static_cast<u8*>(dst.ptr()) + addrOffset);

        CHK_RET(hrtMemAsyncCopy(dstSplit, dst.size(), const_cast<const void*>(srcSplit),
            contSplit, HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE, stream.ptr()));
        u32 taskID = 0;
        u32 streamID = 0;
        hrtGetTaskIdAndStreamID(taskID, streamID);
        HCCL_INFO("[SendRecvExecutor][MemcpyAsyncD2D] MemcpyAsync para: dst[%p] destMax[%llu] src[%p] count[%llu], "\
            "taskID[%u], streamID[%u]", dstSplit, dst.size(),
            const_cast<const void*>(srcSplit), contSplit, taskID, streamID);
    }
    return HCCL_SUCCESS;
}
} // namespace hccl
